package com.cws4cloud.cws.mall.xsm.controller;

/**
 * @ClassName: MessageHandler
 * @Description: 消息处理器
 * @Date 2021/6/19 17:50
 */

import com.cws4cloud.cws.mall.xsm.service.TradelogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * <p>
 * 消息处理器
 * </p>
 *
 * @author yangkai.shen
 * @date Created in 2019-01-07 14:58
 */
@Component
@Slf4j
public class MessageHandler {

	@Autowired
	private TradelogService tradelogService;
	@Autowired
	private RedisTemplate redisTemplate;

	@KafkaListener(topics = "#{'${cws.kafka.topics}'.split(',')}", containerFactory = "ackContainerFactory")
	public void handleMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			Optional<?> kafkaMessage = Optional.ofNullable(record.value());
			if (kafkaMessage.isPresent()) {
				Object message =kafkaMessage.get();
				log.info("收到消息: {}", message);
				boolean isSave = tradelogService.saveTradeLog(message);
				String topic = record.topic();
				if (!isSave){
					log.error("订单消息消费失败，topic:{}-具体内容: {}",topic,message);
				}
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}
}
